package com.wanrue.tcc.commom;

import com.wanrue.tcc.common.TccMethod;
import com.wanrue.tcc.context.*;
import com.wanrue.tcc.enums.Result;
import com.wanrue.tcc.enums.Role;
import com.wanrue.tcc.enums.Status;
import com.wanrue.tcc.exception.CancellingException;
import com.wanrue.tcc.exception.ConfirmingException;
import com.wanrue.tcc.exception.NoExistedTransactionException;
import com.wanrue.tcc.exception.SystemException;
import com.wanrue.tcc.repository.Repository;
import lombok.Setter;
import lombok.extern.log4j.Log4j;

import javax.annotation.Resource;
import java.util.Deque;
import java.util.LinkedList;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * 事务管理器
 * Created by changmingxie on 10/26/15.
 */
@Setter
@Log4j
public class TransactionManager {

    @Resource
    private Repository repository;

    private ExecutorService executorService;

    private static final ThreadLocal<Deque<Transaction>> CURRENT = new ThreadLocal<Deque<Transaction>>();

    public TransactionManager() {
    }

    public void init() {
        if (executorService == null) {
            synchronized (TransactionManager.class) {
                if (executorService == null) {
                    executorService = Executors.newCachedThreadPool();
                }
            }
        }
    }

    public void destroy() {
        if (executorService == null) {
            synchronized (TransactionManager.class) {
                executorService.shutdown();
            }
        }
    }

    /**
     * 开启根事务
     *
     * @param tccMethod tcc方法信息
     * @return
     */
    public Transaction begin(TccMethod tccMethod) {
        Transaction transaction = new Transaction(Role.INITIATOR);
        buidInvocation(transaction, tccMethod);
        return transaction;
    }

    /**
     * 开启分支事务
     *
     * @param context   事务上下文
     * @param tccMethod tcc方法信息
     * @return
     */
    public Transaction beginBranch(Context context, TccMethod tccMethod) {
        Transaction transaction = new Transaction(context);
        buidInvocation(transaction, tccMethod);
        return transaction;
    }

    /**
     * 构建cc调用点信息
     *
     * @param transaction 事务信息
     * @param tccMethod   tcc方法信息
     */
    private void buidInvocation(Transaction transaction, TccMethod tccMethod) {
        Object[] args = tccMethod.getArgs();
        String cancel = tccMethod.getCancel();
        String confirm = tccMethod.getConfirm();
        Class target = tccMethod.getTarget().getClass();
        Class<?>[] parameterTypes = tccMethod.getMethod().getParameterTypes();
        Invocation confirmInvocation = new Invocation(target, confirm, parameterTypes, args);
        Invocation cancelInvocation = new Invocation(target, cancel, parameterTypes, args);
        Xid xid = transaction.getXid();
        Participant participant = new Participant(xid, confirmInvocation, cancelInvocation);
        transaction.addParticipant(participant);
        repository.create(transaction);
        registerTransaction(transaction);
    }

    /**
     * 参与者cc点被调用时获取事务信息
     *
     * @param context 事务上下文
     * @return 事务信息
     * @throws NoExistedTransactionException
     */
    public Transaction getBranch(Context context) throws NoExistedTransactionException {
        Transaction transaction = repository.findByXid(context.getXid());
        if (transaction != null) {
            transaction.setStatus(context.getStatus());
            registerTransaction(transaction);
            return transaction;
        } else {
            throw new NoExistedTransactionException();
        }
    }

    /**
     * 提交事务
     *
     * @param asyncCommit 是否异步调用
     */
    public void commit(boolean asyncCommit) {
        Transaction transaction = getCurrentTransaction();
        transaction.setStatus(Status.CONFIRMING);
        int update = repository.update(transaction, Result.INITIAL);
        if (update <= 0) return;
        if (asyncCommit) {
            try {
                Long statTime = System.currentTimeMillis();
                executorService.submit(() -> commit(transaction));
                log.debug("async submit cost time:" + (System.currentTimeMillis() - statTime));
            } catch (Throwable commitException) {
                log.warn("transaction async submit confirm failed, recovery job will try to confirm later.", commitException);
                throw new ConfirmingException(commitException);
            }
        } else {
            commit(transaction);
        }
    }

    /**
     * 提交事务
     *
     * @param transaction
     */
    private void commit(Transaction transaction) {
        try {
            transaction.commit();
            repository.update(transaction, Result.SUCCESS);
        } catch (Throwable commitException) {
            repository.update(transaction, Result.FAILURE);
            log.warn("transaction confirm failed, recovery job will try to confirm later.", commitException);
            throw new ConfirmingException(commitException);
        }
    }

    /**
     * 回滚事务
     *
     * @param asyncRollback 是否异步调用
     */
    public void rollback(boolean asyncRollback) {
        final Transaction transaction = getCurrentTransaction();
        transaction.setStatus(Status.CANCELLING);
        int update = repository.update(transaction, Result.INITIAL);
        if (update <= 0) return;
        if (asyncRollback) {
            try {
                executorService.submit(() -> rollback(transaction));
            } catch (Throwable rollbackException) {
                log.warn("transaction async rollback failed, recovery job will try to rollback later.", rollbackException);
                throw new CancellingException(rollbackException);
            }
        } else {
            rollback(transaction);
        }
    }

    /**
     * 回滚事务
     *
     * @param transaction
     */
    private void rollback(Transaction transaction) {
        try {
            transaction.rollback();
            repository.update(transaction, Result.SUCCESS);
        } catch (Throwable rollbackException) {
            repository.update(transaction, Result.FAILURE);
            log.warn("transaction rollback failed, recovery job will try to rollback later.", rollbackException);
            throw new CancellingException(rollbackException);
        }
    }

    /**
     * 是否已经开启事务
     *
     * @return
     */
    public boolean isTransactionActive() {
        Deque<Transaction> transactions = CURRENT.get();
        return transactions != null && !transactions.isEmpty();
    }

    /**
     * 得到当前事务
     *
     * @return
     */
    public Transaction getCurrentTransaction() {
        if (isTransactionActive()) {
            return CURRENT.get().peek();
        }
        return null;
    }

    /**
     * 清理资源
     *
     * @param transaction 事务信息
     */
    public void clean(Transaction transaction) {
        if (isTransactionActive() && transaction != null) {
            Transaction currentTransaction = getCurrentTransaction();
            if (currentTransaction == transaction) {
                CURRENT.get().pop();
            } else {
                throw new SystemException("Illegal transaction when clean after completion");
            }
        }
    }

    /**
     * 添加参与者方法信息
     *
     * @param participant 参与者方法信息
     */
    public void addParticipant(Participant participant) {
        Transaction transaction = this.getCurrentTransaction();
        transaction.addParticipant(participant);
        transaction.setVersion(transaction.getParticipants().size() - 1);
        repository.update(transaction);
    }

    /**
     * 注册事务
     *
     * @param transaction 事务信息
     */
    private void registerTransaction(Transaction transaction) {
        if (CURRENT.get() == null) {
            CURRENT.set(new LinkedList<Transaction>());
        }
        CURRENT.get().push(transaction);
    }
}
